package com.xyzq.cgate.core.component.hystrix;

import com.alibaba.fastjson.JSON;
import com.netflix.hystrix.HystrixObservableCommand;
import com.xyzq.cgate.common.constant.Constants;
import com.xyzq.cgate.common.enums.RequestTypeEnum;
import com.xyzq.cgate.common.exception.CgateException;
import com.xyzq.cgate.common.rule.custom.entity.MatchCondition;
import com.xyzq.cgate.core.plugin.IPluginChain;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.ClientResponse;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;
import rx.Observable;
import rx.RxReactiveStreams;
import rx.schedulers.Schedulers;

import java.time.Duration;
import java.util.Optional;

/**
 * @Description:
 * @author chenchen
 * @date 2019/6/25 13:50
 */
public class HttpObservableCommand extends HystrixObservableCommand<Void> {

    private static final Logger LOGGER = LoggerFactory.getLogger(HttpObservableCommand.class);

    private static final WebClient WEB_CLIENT = WebClient.create();

    private final ServerWebExchange exchange;

    private final IPluginChain chain;

    private final MatchCondition condition;

    private final String url;

    private final long retry;

    public HttpObservableCommand(final HystrixObservableCommand.Setter setter,
                       final ServerWebExchange exchange,
                       final IPluginChain chain,
                       final MatchCondition condition,
                       final String url,
                       final long retry) {
        super(setter);
        this.exchange = exchange;
        this.chain = chain;
        this.condition = condition;
        this.url = url;
        this.retry = retry;
    }

    @Override
    protected Observable<Void> construct() {
        return RxReactiveStreams.toObservable(doHttpInvoke());
    }

    private Mono<Void> doHttpInvoke() {
        if(StringUtils.equalsIgnoreCase(RequestTypeEnum.GET.getRequestType(), condition.getQueryType())) {
            return WEB_CLIENT.get().uri(url)
                    .headers(httpHeaders -> {
                        httpHeaders.addAll(exchange.getRequest().getHeaders());
                        httpHeaders.remove(HttpHeaders.HOST);
                    })
                    .exchange()
                    .retry(retry)
                    .doOnError(this::dealError)
                    .flatMap(this::dealResponse);

        } else if(StringUtils.equalsIgnoreCase(RequestTypeEnum.POST.getRequestType(), condition.getQueryType())) {
            return WEB_CLIENT.post().uri(url)
                    .headers(httpHeaders -> {
                        httpHeaders.addAll(exchange.getRequest().getHeaders());
                        httpHeaders.remove(HttpHeaders.HOST);
                    })
                    .contentType(buildMediaType())
                    .body(BodyInserters.fromDataBuffers(exchange.getRequest().getBody()))
                    .exchange()
                    .doOnError(this::dealError)
                    .flatMap(this::dealResponse);

        }
        return Mono.empty();
    }

    private Mono<Void> dealResponse(ClientResponse clientResponse) {
        HttpStatus httpStatus = clientResponse.statusCode();
        LOGGER.debug("-----------请求完成----------");
        if (httpStatus.is2xxSuccessful()) {
            exchange.getResponse().setStatusCode(HttpStatus.OK);
            exchange.getAttributes().put(Constants.WEBCLIENT_RESPONSE, clientResponse);
        } else {
            throw new CgateException(httpStatus.getReasonPhrase());
        }
        return chain.execute(exchange);
    }

    @Override
    public Observable<Void> resumeWithFallback() {
        ServerHttpResponse response = exchange.getResponse();
        return RxReactiveStreams.toObservable(
                response.writeWith(
                        Mono.just(response
                                .bufferFactory()
                                .wrap(JSON.toJSONBytes(
                                        HttpStatus.INTERNAL_SERVER_ERROR.getReasonPhrase())))));
    }

    private MediaType buildMediaType() {
        return MediaType.valueOf(Optional.ofNullable(exchange
                .getRequest()
                .getHeaders().getFirst(HttpHeaders.CONTENT_TYPE))
                .orElse(MediaType.APPLICATION_JSON_UTF8_VALUE));
    }

    private void dealError(Throwable e) {
        e.printStackTrace();
        throw new CgateException("webclient call error");
    }
}
